package com.shujia.spark.core

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD

object Demo27HomeWork {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()

    conf.setMaster("local")
    conf.setAppName("hw")

    val sc = new SparkContext(conf)

    val subjectRDD: RDD[String] = sc.textFile("data/subject.txt")
    val scoreRDD: RDD[String] = sc.textFile("data/score.txt")

    /**
     * 1 关联科目表获取科目总分
     *
     */
    val subKVRDD: RDD[(String, String)] = subjectRDD.map(line => {
      val split: Array[String] = line.split(",")
      (split(0), line)
    })

    val scoKVRDD: RDD[(String, String)] = scoreRDD.map(line => {
      val split: Array[String] = line.split(",")
      (split(1), line)
    })
    val joinRDD: RDD[(String, (String, String))] = subKVRDD.join(scoKVRDD)

    //整理数据
    val scoreAndSubRDD: RDD[(String, String, Double, Double)] = joinRDD.map {
      case (cId: String, (subjectInfo: String, sccoInfo: String)) =>
        val subSplit: Array[String] = subjectInfo.split(",")
        //科目总分
        val subjevtScore: Double = subSplit(2).toDouble
        //科目名
        val cname: String = subSplit(1)

        val scoSplit: Array[String] = sccoInfo.split(",")
        //学号
        val id: String = scoSplit(0)
        //科目分数
        val score: Double = scoSplit(2).toDouble
        (id, cname, score, subjevtScore)
    }

    /**
     * 2 对分数做归一化, 除以总分
     */
    val guiYiRDD: RDD[(String, Double)] = scoreAndSubRDD.map {
      case (id, cname, score, subjevtScore) =>
        (id, score / subjevtScore)
    }

    /**
     * 3  按照学号分数,计算方差
     */
    //按照学号分组
    val groupByRDD: RDD[(String, Iterable[Double])] = guiYiRDD.groupByKey()
    val stdRDD: RDD[(String, Double)] = groupByRDD.map {
      case (id: String, scos: Iterable[Double]) =>

        val scoList: List[Double] = scos.toList

        //1 计算平均数
        val avg: Double = scoList.sum / scoList.length

        // 计算分数和平均分差值平方和, 计算方差
        val std: Double = scoList.map(sco => (sco - avg) * (sco - avg)).sum / scoList.length

        (id, std)
    }

    /**
     * 4 按照方差降序排序,取前100
     */
    val ids: Array[String] = stdRDD
      .sortBy(kv => kv._2, ascending = false)
      .take(2)
      .map(kv => kv._1)

    /**
     * 5 获取前100学生各科的分数
     */
    val resultRDD: RDD[(String, String, Double, Double)] = scoreAndSubRDD.filter {
      case (id, cname, score, subjevtScore) =>
        ids.contains(id)
    }

    resultRDD.sortBy(_._1).foreach(println)

  }

}
